package com.app.kafka

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._


object Process {

  def processData(ssc: StreamingContext): Unit = {

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "hello-group",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("topicA")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val kstream: DStream[(String, Int)] = stream.transform { record =>
      val rdd: RDD[(String, Int)] = record.filter { record =>
        val message: String = record.value()
        null != message && message.trim.split(",").length == 4
      }.map { record =>
        val keyword: String = record.value().trim.split(",").last
        (keyword, 1)
      }.reduceByKey(_ + _)
      rdd
    }

    kstream.print()
  }
}